package com.ds.infrastructure.hyperspace.container.endpoint.entry;

import com.google.common.net.InetAddresses;
import com.ds.infrastructure.hyperspace.common.EndpointProtocolEnum;
import com.ds.infrastructure.hyperspace.common.TargetEndpointStream;
import com.ds.infrastructure.hyperspace.common.config.EntryNodeConfig;
import com.ds.infrastructure.hyperspace.common.flow.EntryProtocolAdapter;
import com.ds.infrastructure.hyperspace.container.endpoint.connector.HyperspaceConnector;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.operation.EntryFlowOperation;
import com.ds.infrastructure.hyperspace.container.internal.NettyContainer;
import io.netty.channel.*;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.net.SocketAddress;

import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.COLON;
import static com.ds.infrastructure.hyperspace.container.internal.util.StringConstant.HYPHEN;

/**
 * @author  Weishaoying
 */
@Slf4j
public abstract class AbstractEntryProtocolAdapter extends ChannelDuplexHandler
        implements EntryProtocolAdapter<HyperspaceDuplexFlowContext> {

    protected final NettyContainer nettyContainer;

    protected final EntryNodeConfig entryNodeConfig;

    private final String entryNodeInfo;

    protected final EntryFlowOperation<HyperspaceDuplexFlowContext> entryFlowOperation;

    protected ChannelHandlerContext channelHandlerContext;

    protected Channel channel;

    @Getter
    private EventLoop entryEventLoop;

    protected HyperspaceConnector hyperspaceConnector;

    protected String clientIp = HYPHEN;

    private boolean isOpenChannel = true;

    private boolean fireChannelInactive = false;

    public AbstractEntryProtocolAdapter(NettyContainer nettyContainer, EntryNodeConfig entryNodeConfig) {
        this.nettyContainer = nettyContainer;
        this.entryNodeConfig = entryNodeConfig;
        this.entryNodeInfo = new StringBuilder().append(entryNodeConfig.getHost()).append(COLON)
                .append(entryNodeConfig.getPort()).toString();
        this.entryFlowOperation = nettyContainer.getFlowPipelineOrchestrator().getEntryFlowOperation();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        resetChannelActiveState(ctx, null);
        entryFlowOperation.handleChannelActive(entryNodeInfo);
        ctx.fireChannelActive();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        try {
            if (!fireChannelInactive) {
                fireChannelInactive = true;
                entryFlowOperation.handleChannelInactive(entryNodeInfo, currentDuplexFlowContext());
                releaseDuplexFlowContext();
            }
        } catch (Exception ex) {
            log.error("Occur an exception when call channel inactive method! channel id {}",getChannelId(), ex);
        } finally {
            ctx.fireChannelInactive();
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            handleChannelRead(ctx, msg);
        } catch (Exception ex) {
            releaseData(msg);
            closeEntryChannel();
            throw ex;
        }
    }

    /**
     * Is called for each message
     *
     * @param ctx
     * @param msg
     * @throws Exception
     */
    protected abstract void handleChannelRead(ChannelHandlerContext ctx, Object msg);

    @Override
    public void flushToEntry() {
        channelHandlerContext.flush();
    }

    @Override
    public void interruptToEntry(HyperspaceDuplexFlowContext duplexFlowContext) {
        resetChannelCodecState(duplexFlowContext);
        if (duplexFlowContext.getEndpointProtocol() != EndpointProtocolEnum.HTTP2
                && duplexFlowContext.getEndpointProtocol() != EndpointProtocolEnum.HTTP2C) {
            ChannelOutboundBuffer channelOutboundBuffer = channel.unsafe().outboundBuffer();
            if (channelOutboundBuffer != null) {
                channelOutboundBuffer.addFlush();
                for (; ; ) {
                    if (!channelOutboundBuffer.remove()) {
                        break;
                    }
                }
            }
        }
        interruptToEntryComplete(duplexFlowContext);
    }

    public void resetChannelActiveState(ChannelHandlerContext ctx,
                                        HyperspaceDuplexFlowContext duplexFlowContext) {
        this.channelHandlerContext = ctx;
        this.channel = ctx.channel();
        this.entryEventLoop = channel.eventLoop();
        this.hyperspaceConnector = nettyContainer.getHyperspaceConnector(entryEventLoop);
        SocketAddress remoteAddress = channel.remoteAddress();
        if (remoteAddress instanceof InetSocketAddress) {
            InetSocketAddress inetSocketAddress = (InetSocketAddress) remoteAddress;
            if (inetSocketAddress.getAddress() != null) {
                this.clientIp = InetAddresses.toAddrString(inetSocketAddress.getAddress());
            }
        }
        if (duplexFlowContext != null) {
            duplexFlowContext.setEntryProtocolAdapter(this);
        }
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) {
        HyperspaceDuplexFlowContext duplexFlowContext = currentDuplexFlowContext();
        if (duplexFlowContext == null ||
                duplexFlowContext.getCurrentReferTargetEndpointStream() == null) {
            return;
        }

        boolean isChanged = false;
        TargetEndpointStream targetEndpointStream = duplexFlowContext.getCurrentReferTargetEndpointStream();
        try {
            if (!channel.isWritable()) {
                isChanged = true;
                targetEndpointStream.changeTargetChannelReadable(false);
                ctx.flush();
            }
        } finally {
            if (isChanged) {
                targetEndpointStream.changeTargetChannelReadable(true);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        String errorMessage = String.format(
                "Occur an exception that entry endpoint handler unexpected, client ip %s,channel id %s !cause by %s",
                clientIp, getChannelId(), cause);
        log.error(errorMessage, cause);
        entryFlowOperation.handleExceptionCaught(currentDuplexFlowContext(), cause);
    }

    @Override
    public void closeEntryChannel() {
        if (isOpenChannel && channel.isOpen()) {
            isOpenChannel = false;
            channel.flush().close();
        }
    }

    @Override
    public HyperspaceDuplexFlowContext requireDuplexFlowContext(int flowContextKey) {
        // no implements
        return null;
    }

    /**
     * Reset codec state for next codec
     *
     * @param duplexFlowContext
     */
    protected void resetChannelCodecState(HyperspaceDuplexFlowContext duplexFlowContext) {
        // no implements
    }

    /**
     * Response interrupt data to entry
     *
     * @param duplexFlowContext
     */
    protected abstract void interruptToEntryComplete(HyperspaceDuplexFlowContext duplexFlowContext);

    @Override
    public boolean isChannelOpen() {
        return channel.isOpen();
    }

    private String getChannelId() {
        if (channel == null) {
            return HYPHEN;
        }
        return String.valueOf(channel.id());
    }

}
